/*
    Copyright 2009 Barnet Wagman

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package ot.s3;

import ot.*;
import java.util.*;
import java.io.*;
import java.util.ArrayList;
import org.jets3t.service.*;

/** 'Managed' versions of the operations in {@link OpsS3 OpsS3}.  When
 *  an <tt>OpsS3</tt> method thows and exception, methods in this
 *  class record the exception and retry the operation up to a specified
 *  number of times, at a specified interval.  The max number of tries
 *  and the interval between then can be reset either in this class or
 *  in <tt>PojoS3</tt>.
 *
 * @author Barnet Wagman
 */
public class ManagedOps {

    public OpsS3 ops;
//    Serializer serializer;
    SerializerTable sert;
    ErrorHandler errorHandler;      
    
    public ManagedOps(OpsS3 ops, SerializerTable sert,
                      ErrorHandler errorHandler) {
        this.ops = ops;
        this.errorHandler = errorHandler;
        this.sert = sert;
    }
    
        /** Uses {@link LimitedRetryErrorHandler} as the
         *  error handler, with parameteres <tt>
         *  ConstantsS3.MAX_TRIES_DEFAULT</tt> and <tt>
            ConstantsS3.RETRY_INTERVAL_MILLIS_DEFAULT</tt>.
         * 
         * @param ops
         */
     public ManagedOps(OpsS3 ops,SerializerTable sert) {
        this.sert = sert;
        this.ops = ops;
        this.errorHandler =
             new LimitedRetryErrorHandler(ConstantsOtS3.MAX_TRIES_DEFAULT,
                                          ConstantsOtS3.RETRY_INTERVAL_MILLIS_DEFAULT);
     }
    
        // ------------- List objects  ----------------------
            
    public List<String> listS3Objects(String bucketName) 
        throws OpFailedException {
        
        ErrorHandler eh = errorHandler.cloneErrorHandler(); 
        
        for (;;) {
            try {
                return( ops.listS3Objects(bucketName) );
            }
            catch(Throwable xxx) {
                quitOrRetryReadOp(eh, xxx," bucketName=" + bucketName);                                                 
            }
        }        
    }

    public List<String> listS3Objects(LocD loc)
        throws OpFailedException {

        ErrorHandler eh = errorHandler.cloneErrorHandler();
        
        for (;;) {
            try {
                return( ops.listS3Objects(
                            ((ContextS3) loc.context).bucketName,
                            loc.getPath().toPathString()) );
            }
            catch(Throwable xxx) {
                quitOrRetryReadOp(eh, xxx," bucketName=" +
                                    ((ContextS3) loc.context).bucketName +
                     " pathString=" + loc.getPath().toPathString());
            }
        }
    }
    
        // ---------- Get operations ---------------------------------
    
    public byte[] getS3ObjBytes(String bucketName,String key) 
        throws OpFailedException {
        
        ErrorHandler eh = errorHandler.cloneErrorHandler(); 
        
        List<Throwable> errs = null;
        int nTries = 0;

        for ( ;; ) {
            try {
                return( ops.getS3ObjBytes(bucketName,key) );
            }
            catch(Throwable xxx) {
                quitOrRetryReadOp(eh, xxx,
                                  " bucketName=" + bucketName + 
                                  " S3 key=" + key);                 
                System.out.println("Retrying after catching exception: " +
                                   xxx.getMessage());
            }
        }               
    }

    public Serializable getS3Obj(LocO loc) throws OtException {
        GetS3ObjBytesTask t = new GetS3ObjBytesTask(loc);
        t.run();
        return( (Serializable)
                sert.getSer(loc).byteArrayToObject(t.retrievedBytes) );
//                serializer.byteArrayToObject(t.retrievedBytes) );
    }

    public List<Serializable> getS3Objs(List<LocO> locs) 
        throws OpFailedException, OtException {

//        List<byte[]> bl = getS3ObjBytes(locs);
        List<GetS3ObjBytesTask> bls = getS3ObjBytes(locs);
        List<Serializable> objs = new ArrayList<Serializable>();

//        for ( byte[] b : bl ) {
        for ( GetS3ObjBytesTask bl: bls ) {
//            if ( b != null )
            if ( bl.getRetrievedBytes() != null ) {
                objs.add((Serializable) 
                          sert.getSer(bl.loc).byteArrayToObject(bl.retrievedBytes));
//                objs.add((Serializable) serializer.byteArrayToObject(b));
            }
        }
        return(objs);
    }

//    synchronized public List<byte[]> getS3ObjBytes(List<LocO> locs)
    synchronized public List<GetS3ObjBytesTask> getS3ObjBytes(List<LocO> locs)
        throws OpFailedException {
        
        int maxThreads = ops.getMaxNThreads();
        
        List<GetS3ObjBytesTask> tasks = new ArrayList<GetS3ObjBytesTask>();
        
        for ( LocO loc : locs ) {
            GetS3ObjBytesTask t = new GetS3ObjBytesTask(loc);
            (new Thread(t)).start();            
            tasks.add(t);
            while ( nRunningGetTasks(tasks) >= maxThreads ) UtilOtS3.nap(100);
        }
        while ( nRunningGetTasks(tasks) > 0 ) UtilOtS3.nap(100);
        
            // Get results and check for failures
        int nFailed = 0;
//        List<byte[]> rList = new ArrayList<byte[]>();
        List<GetS3ObjBytesTask> rList = new ArrayList<GetS3ObjBytesTask>();
        for ( GetS3ObjBytesTask t : tasks ) {
            if ( !t.finishedSuccessfully ) ++nFailed;
            else rList.add(t);
//            else rList.add(t.getRetrievedBytes());
        }
        if ( nFailed < 1 ) { return(rList); }
        else {   
            String mess = nFailed + " failures occurred attempting to read " +
                    locs.size() + " PojoSets. Paths of failures:\n";
            List<Throwable> errs = new ArrayList<Throwable>();
            for ( GetS3ObjBytesTask t : tasks ) {
                if ( !t.finishedSuccessfully ) {
                    mess += t.loc.toString();
                    if ( t.errs != null ) errs.addAll(t.errs);
                }
            }
            OpFailedException ox = new OpFailedException(mess);
            ox.setThrowableList(errs);
            throw ox;
        }        
    }
    
    int nRunningGetTasks(List<GetS3ObjBytesTask> tasks) {
        
        int n = 0;
        for ( GetS3ObjBytesTask t : tasks ) {
            if ( !t.isDone ) ++n;
        }
        return(n);
    }    
    
    class GetS3ObjBytesTask implements Runnable {
        LocO loc;
        byte[] retrievedBytes;
        boolean isDone;
        boolean finishedSuccessfully;
        List<Throwable> errs;
        
        GetS3ObjBytesTask(LocO loc) {
            this.loc = loc;
        }
        
        public void run() {
            finishedSuccessfully = false;
            ErrorHandler eh = errorHandler.cloneErrorHandler();

            for (;;) {
                try {                    
                    retrievedBytes =
                            ops.getS3ObjBytes(
                                ((ContextS3) loc.context).bucketName,
                                loc.getPath().toPathString());
                    isDone = true;
                    finishedSuccessfully = true;
                    return;
                }
                catch(Throwable xxx) {      
                    if ( errs == null ) errs = new ArrayList<Throwable>();
                    errs.add(xxx);
                    ErrorHandler.Action a = eh.add(xxx);
                    if ( a.equals(ErrorHandler.Action.quit) ) {
                        isDone = true;
                        finishedSuccessfully = false;
                        return;                    
                    }                
                    else if ( a.equals(ErrorHandler.Action.retry)) {     
                        for (;;) {
                            try {
                                replaceReadService();
                            }
                            catch(OpFailedException ox) {
                                errs.add(ox);
                                a = eh.add(ox);
                            }
                            if ( a.equals(ErrorHandler.Action.quit) ) {
                                isDone = true;
                                finishedSuccessfully = false;
                                return;
                            }                            
                        }                                       
                    }
                    else {
                        isDone = true;
                        finishedSuccessfully = false;  
                        throw new RuntimeException("Undefined action=" + a);
                    }                      
                }
            }                                        
        }
        
        byte[] getRetrievedBytes() { return(retrievedBytes); }
    }
    
        // ---------------- Bucket operations  ----------------
    
    public List<String> listBuckets() throws OpFailedException {
        
        ErrorHandler eh = errorHandler.cloneErrorHandler(); 
        
        for (;; ) {
            try {
                return( ops.listBuckets() );
            }
            catch(Throwable xxx) {
                quitOrRetryReadOp(eh, xxx, "");                
            }
        }
    }
    
        /** @return true if successful, false if the bucket already exists.
         * 
         * @param bucketName
         * @return
         * @throws org.jets3t.service.S3ServiceException
         */
    public boolean createBucket(String bucketName) throws OpFailedException {
        
        ErrorHandler eh = errorHandler.cloneErrorHandler(); 
        
        List<Throwable> errs = null;

        for ( ;; ) {
            try {
                return( ops.createBucket(bucketName) );
            }
            catch(Throwable xxx) {
                quitOrRetryWriteOp(eh, xxx,"bucket=" + bucketName);                                              
            }
        }        
    }
    
        /** @return true if successful, false if the bucket does not exist
         * 
         * @param bucketName
         * @return
         * @throws org.jets3t.service.S3ServiceException
         */
    public boolean deleteBucket(String bucketName) throws OpFailedException {
       
       ErrorHandler eh = errorHandler.cloneErrorHandler();
        
        for (;;) {    
            try {                
                return( ops.deleteBucket(bucketName) );
            }
            catch(Throwable xxx) {
                quitOrRetryWriteOp(eh, xxx,"bucket=" + bucketName);                         
            }
        }        
    }
    
    public boolean bucketExists(String bucketName) throws OpFailedException {
       
        ErrorHandler eh = errorHandler.cloneErrorHandler();
        
        for (;;) {        
            try {
                return( ops.bucketExists(bucketName) );
            }
            catch(Throwable xxx) {
                quitOrRetryReadOp(eh, xxx,"bucket=" + bucketName);                              
            }
        }        
    }
    
        // ---------- Delete object -------------------------------
    
    public void deleteS3Object(String bucketName,String key) throws OpFailedException {
        
        ErrorHandler eh = errorHandler.cloneErrorHandler();
        
        for (;;) {        
            try {
                ops.deleteS3Object(bucketName, key);
                return;
            }
            catch(Throwable xxx) {            
                quitOrRetryWriteOp(eh, xxx,"bucket=" + bucketName +
                                           " S3 key=" + key);                          
            }    
        }
    }
      
        // ------------ Put operations ---------------------------

    synchronized public void putObjToS3(LocO loc, Serializable obj) {
        PutS3ObjBytesTask t = new PutS3ObjBytesTask(loc, obj);
        t.run();
    }
    
    synchronized public void putS3Objs(List<LocO> locs,
                                       List<Serializable> objs) throws OpFailedException{

        int maxThreads = ops.getMaxNThreads();

        List<PutS3ObjBytesTask> tasks = new ArrayList<PutS3ObjBytesTask>();

        for ( int i = 0;i < locs.size(); i++ ) {
            PutS3ObjBytesTask t = 
                new PutS3ObjBytesTask(locs.get(i),objs.get(i));
            (new Thread(t)).start();
            tasks.add(t);
            while ( nRunningPutTasks(tasks) >= maxThreads ) UtilOtS3.nap(100);
        }
        while ( nRunningPutTasks(tasks) > 0 ) UtilOtS3.nap(100);

            // Check for failures
        int nFailed = 0;
        for ( PutS3ObjBytesTask t : tasks ) {
            if ( !t.finishedSuccessfully ) ++nFailed;
        }
        if ( nFailed < 1 ) return;
        else {
            String mess = nFailed + " failures occurred attempting to write " +
                    objs.size() + " PojoSets. Paths of failures:\n";
            List<Throwable> errs = new ArrayList<Throwable>();
            for ( PutS3ObjBytesTask t : tasks ) {
                if ( !t.finishedSuccessfully ) {
                    mess += t.loc.toString();
                    if ( t.errs != null ) errs.addAll(t.errs);
                }
            }
            OpFailedException ox = new OpFailedException(mess);
            ox.setThrowableList(errs);
            throw ox;
        }
    }
       
    int nRunningPutTasks(List<PutS3ObjBytesTask> tasks) {

        int n = 0;
        for ( PutS3ObjBytesTask t : tasks ) {
            if ( !t.isDone ) ++n;
        }
        return(n);
    }
    
    class PutS3ObjBytesTask implements Runnable {

        LocO loc;
        Serializable obj;

        boolean isDone;
        boolean finishedSuccessfully;
        List<Throwable> errs;

        PutS3ObjBytesTask(LocO loc, Serializable obj) {
            this.loc = loc;
            this.obj = obj;
        }

        public void run() {
            isDone = false;
            finishedSuccessfully = false;

            ErrorHandler eh = errorHandler.cloneErrorHandler();

            byte[] objBytes = null;
            for (;;) {
                try {
                    try {
//                        objBytes = serializer.objectToByteArray(obj);
                         objBytes = sert.getSer(loc).objectToByteArray(obj);
                    }
                    catch(Exception zzz) {
                        isDone = true;
                        finishedSuccessfully = false;
                        if ( errs == null ) errs = new ArrayList<Throwable>();
                        errs.add(zzz);
                        OpFailedException ox =
                            new OpFailedException("Exception was thrown serializing " +
                                                   loc.toString());
                        errs.add(ox);
                        return;
                    }
                    ops.putS3ObjBytes(((ContextS3) loc.context).bucketName,
                                      loc.getPath().toPathString(),
                                      objBytes);
                    isDone = true;
                    finishedSuccessfully = true;
                    objBytes = null;
                    obj = null;
                    return;
                }
                catch(Throwable xxx) {
                    if ( errs == null ) errs = new ArrayList<Throwable>();
                    errs.add(xxx);
                    ErrorHandler.Action a = eh.add(xxx);
                    if ( a.equals(ErrorHandler.Action.quit) ) {
                        isDone = true;
                        finishedSuccessfully = false;
                        objBytes = null;
                        obj = null;
                        return;
                    }
                    else if ( a.equals(ErrorHandler.Action.retry)) {
                        for (;;) {
                            try {
                                replaceWriteService();
                            }
                            catch(OpFailedException ox) {
                                errs.add(ox);
                                a = eh.add(ox);
                            }
                            if ( a.equals(ErrorHandler.Action.quit) ) {
                                isDone = true;
                                finishedSuccessfully = false;
                                return;
                            }
                        }
                    }
                    else {
                        isDone = true;
                        finishedSuccessfully = false;
                        throw new RuntimeException("Undefined action=" + a);
                    }
                }
            }
        }
    }        
    
        // ----------- Supporting methods -------------------
    
    private void replaceReadService() throws OpFailedException {        
        try {
            synchronized(ops) {        
                ops.replaceReadService();
            }
        }
        catch(S3ServiceException ex) {
            OpFailedException ox = new OpFailedException("replaceReadService() failed");
            List<Throwable> xs = new ArrayList<Throwable>();
            ox.setThrowableList(xs);
            throw ox;
        }
    }
    
    private void replaceWriteService() throws OpFailedException {
        
        try {
            synchronized(ops) {        
                ops.replaceWriteService();
            }
        }
        catch(S3ServiceException ex) {
            OpFailedException ox = new OpFailedException("replaceWriteService() failed");
            List<Throwable> xs = new ArrayList<Throwable>();
            ox.setThrowableList(xs);
            throw ox;
        }
    }
    
    
    enum ServiceType { read, write };
    
    private void quitOrRetryReadOp(ErrorHandler eh,Throwable x, String message) 
        throws OpFailedException {
        quitOrRetry(eh,x,message,ServiceType.read);
    }
    
    private void quitOrRetryWriteOp(ErrorHandler eh,Throwable x, String message) 
        throws OpFailedException {
        quitOrRetry(eh,x,message,ServiceType.write);
    }
    
    private void quitOrRetry(ErrorHandler eh,Throwable x, String message,
                             ServiceType serviceType)
        throws OpFailedException {

        ErrorHandler.Action a = eh.add(x);
        if ( a.equals(ErrorHandler.Action.retry) ) {
            switch (serviceType) {
                case read:            
                    replaceReadService();
                    System.out.println("Retry -> replaceReadService() after " +
                                       getMessage(x));
                    break;
                case write:
                    replaceWriteService();
                    System.out.println("Retry -> replaceWriteService() after " +
                                       getMessage(x));
                    break;
                default:
                    throw new RuntimeException("Undefined service type=" +
                                               serviceType);
            }
        }
        else if ( a.equals(ErrorHandler.Action.quit) ) {            
            throw createOpFailedException(eh, x, message);
        }
        else throw new RuntimeException("Undefined action=" + a);
    }

    String getMessage(Throwable x) {
        if ( x == null ) return("null Throwable");
        Throwable c = x.getCause();
        if ( c != null ) return(c.getMessage());
        else return( x.getMessage() );
    }
   
    OpFailedException createOpFailedException(ErrorHandler eh,
                                              Throwable x, String message) {
        List<Throwable> xs = eh.getThrowables();
        if ( (xs == null) || (xs.size() < 1) ) {
            return(new OpFailedException(
                    UtilOtS3.getParentMethodNameFullyQualified() + " failed.  " +
                    message));
        }
        else {
            Throwable lastEx = xs.get(xs.size() - 1);
            if ( !(lastEx instanceof S3ServiceException) ) {
                OpFailedException ox = 
                    new OpFailedException(
                        UtilOtS3.getParentMethodNameFullyQualified() + " failed.  " +
                        message);
                ox.setThrowableList(eh.getThrowables());
                return(ox);
            }
            else {
                S3ServiceException sx = (S3ServiceException) lastEx;
                String errCode = sx.getS3ErrorCode();
                if ( errCode != null ) {
                    if ( errCode.equals(ConstantsOtS3.S3_ERROR_CODE_NO_SUCH_KEY) ) {
                        NoSuchS3KeyException kx = 
                            new NoSuchS3KeyException("According to S3, the specfied " +
                                " key does not exist in the specified bucket: " +
                                message);
                        kx.setThrowableList(xs);
                        return(kx);
                    }                
                    else if ( errCode.equals(ConstantsOtS3.S3_ERROR_CODE_NO_SUCH_BUCKET) ) {
                        NoSuchBucketException bx = 
                            new NoSuchBucketException("According to S3, the specified " +
                                " bucket does not exist: " + message);
                        bx.setThrowableList(xs);
                        return(bx);
                    }
                    else {
                        OpFailedException ox = 
                        new OpFailedException(
                            UtilOtS3.getParentMethodNameFullyQualified() + " failed.  " +
                            message);
                        ox.setThrowableList(eh.getThrowables());
                        return(ox);
                    }
                }                
                else {
                    OpFailedException ox = 
                        new OpFailedException(
                            UtilOtS3.getParentMethodNameFullyQualified() + " failed.  " +
                            message);
                    ox.setThrowableList(eh.getThrowables());
                    return(ox);
                }
            }
        }        
    }
}
